package cachedata;

import java.io.EOFException;
import java.io.IOException;
import java.net.InetAddress;
import java.net.SocketTimeoutException;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.util.Iterator;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock.WriteLock;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import buffer.SocketBufferHandler;
import cache.NioSelectorPool;
import thread.Poller;

public class NioSocketWrapper{
	
	  private static final Logger log = LoggerFactory.getLogger(NioSocketWrapper.class);
	  
	  private final NioChannel socket;

	    // Volatile because I/O and setting the timeout values occurs on a different
	    // thread to the thread checking the timeout.
	    private volatile long readTimeout = -1;
	    private volatile long writeTimeout = -1;
	    
	    protected final LinkedBlockingDeque<ByteBufferHolder> bufferedWrites = new LinkedBlockingDeque<>();

	    
	    public long getWriteTimeout() {
			return writeTimeout;
		}

		public void setWriteTimeout(long writeTimeout) {
			this.writeTimeout = writeTimeout;
		}

		private volatile int keepAliveLeft = 100;
	    private volatile boolean upgraded = false;
	    private boolean secure = false;
	    private String negotiatedProtocol = null;
	    /*
	     * Following cached for speed / reduced GC
	     */
	    protected String localAddr = null;
	    protected String localName = null;
	    protected int localPort = -1;
	    protected String remoteAddr = null;
	    protected String remoteHost = null;
	    protected int remotePort = -1;
	    /*
	     * Used if block/non-blocking is set at the socket level. The client is
	     * responsible for the thread-safe use of this field via the locks provided.
	     */
	    private volatile boolean blockingStatus = true;
	    private final Lock blockingStatusReadLock;
	    private final WriteLock blockingStatusWriteLock;
	    /*
	     * Used to record the first IOException that occurs during non-blocking
	     * read/writes that can't be usefully propagated up the stack since there is
	     * no user code or appropriate container code in the stack to handle it.
	     */
	    private volatile IOException error = null;

	    /**
	     * The buffers used for communicating with the socket.
	     */
	    private volatile SocketBufferHandler socketBufferHandler = null;

	    /**
	     * The max size of the buffered write buffer
	     */
	    private int bufferedWriteSize = 64 * 1024; // 64k default write buffer

    private final NioSelectorPool pool;

    private Poller poller = null;
    private int interestOps = 0;
    private CountDownLatch readLatch = null;
    private CountDownLatch writeLatch = null;
    private volatile long lastRead = System.currentTimeMillis();
    private volatile long lastWrite = lastRead;
    
    public NioSocketWrapper(NioChannel channel, NioSelectorPool pool) {
        this.socket = channel;
        ReentrantReadWriteLock lock = new ReentrantReadWriteLock();
        this.blockingStatusReadLock = lock.readLock();
        this.blockingStatusWriteLock = lock.writeLock();
        this.pool = pool;
        socketBufferHandler = channel.getBufHandler();
    }

    public Poller getPoller() { return poller;}
    public void setPoller(Poller poller){this.poller = poller;}
    public int interestOps() { return interestOps;}
    public int interestOps(int ops) { this.interestOps  = ops; return ops; }
    public CountDownLatch getReadLatch() { return readLatch; }
    public CountDownLatch getWriteLatch() { return writeLatch; }
    protected CountDownLatch resetLatch(CountDownLatch latch) {
        if ( latch==null || latch.getCount() == 0 ) return null;
        else throw new IllegalStateException("Latch must be at count 0");
    }
    public void resetReadLatch() { readLatch = resetLatch(readLatch); }
    public void resetWriteLatch() { writeLatch = resetLatch(writeLatch); }

    protected CountDownLatch startLatch(CountDownLatch latch, int cnt) {
        if ( latch == null || latch.getCount() == 0 ) {
            return new CountDownLatch(cnt);
        }
        else throw new IllegalStateException("Latch must be at count 0 or null.");
    }
    public void startReadLatch(int cnt) { readLatch = startLatch(readLatch,cnt);}
    public void startWriteLatch(int cnt) { writeLatch = startLatch(writeLatch,cnt);}

    protected void awaitLatch(CountDownLatch latch, long timeout, TimeUnit unit) throws InterruptedException {
        if ( latch == null ) throw new IllegalStateException("Latch cannot be null");
        // Note: While the return value is ignored if the latch does time
        //       out, logic further up the call stack will trigger a
        //       SocketTimeoutException
        latch.await(timeout,unit);
    }
    public void awaitReadLatch(long timeout, TimeUnit unit) throws InterruptedException { awaitLatch(readLatch,timeout,unit);}
    public void awaitWriteLatch(long timeout, TimeUnit unit) throws InterruptedException { awaitLatch(writeLatch,timeout,unit);}


    public void updateLastWrite() { lastWrite = System.currentTimeMillis(); }
    public long getLastWrite() { return lastWrite; }
    public void updateLastRead() { lastRead = System.currentTimeMillis(); }
    public long getLastRead() { return lastRead; }


    public boolean isReadyForRead() throws IOException {
        socketBufferHandler.configureReadBufferForRead();

        if (socketBufferHandler.getReadBuffer().remaining() > 0) {
            return true;
        }

        fillReadBuffer(false);

        boolean isReady = socketBufferHandler.getReadBuffer().position() > 0;
        return isReady;
    }


    public int read(boolean block, byte[] b, int off, int len) throws IOException {
        int nRead = populateReadBuffer(b, off, len);
        if (nRead > 0) {
            return nRead;
            /*
             * Since more bytes may have arrived since the buffer was last
             * filled, it is an option at this point to perform a
             * non-blocking read. However correctly handling the case if
             * that read returns end of stream adds complexity. Therefore,
             * at the moment, the preference is for simplicity.
             */
        }

        // Fill the read buffer as best we can.
        nRead = fillReadBuffer(block);
        updateLastRead();

        // Fill as much of the remaining byte array as possible with the
        // data that was just read
        if (nRead > 0) {
            socketBufferHandler.configureReadBufferForRead();
            nRead = Math.min(nRead, len);
            socketBufferHandler.getReadBuffer().get(b, off, nRead);
        }
        return nRead;
    }


    public int read(boolean block, ByteBuffer to) throws IOException {
        int nRead = populateReadBuffer(to);
        if (nRead > 0) {
            return nRead;
            /*
             * Since more bytes may have arrived since the buffer was last
             * filled, it is an option at this point to perform a
             * non-blocking read. However correctly handling the case if
             * that read returns end of stream adds complexity. Therefore,
             * at the moment, the preference is for simplicity.
             */
        }

        // The socket read buffer capacity is socket.appReadBufSize
        int limit = socketBufferHandler.getReadBuffer().capacity();
        if (to.remaining() >= limit) {
            to.limit(to.position() + limit);
            nRead = fillReadBuffer(block, to);
            updateLastRead();
        } else {
            // Fill the read buffer as best we can.
            nRead = fillReadBuffer(block);
            updateLastRead();

            // Fill as much of the remaining byte array as possible with the
            // data that was just read
            if (nRead > 0) {
                nRead = populateReadBuffer(to);
            }
        }
        return nRead;
    }
    
    protected int populateReadBuffer(ByteBuffer to) {
        // Is there enough data in the read buffer to satisfy this request?
        // Copy what data there is in the read buffer to the byte array
        socketBufferHandler.configureReadBufferForRead();
        int nRead = transfer(socketBufferHandler.getReadBuffer(), to);

        if (log.isDebugEnabled()) {
            log.debug("Socket: [" + this + "], Read from buffer: [" + nRead + "]");
        }
        return nRead;
    }
    
    private int populateReadBuffer(byte[] b, int off, int len) {
        socketBufferHandler.configureReadBufferForRead();
        ByteBuffer readBuffer = socketBufferHandler.getReadBuffer();
        int remaining = readBuffer.remaining();

        // Is there enough data in the read buffer to satisfy this request?
        // Copy what data there is in the read buffer to the byte array
        if (remaining > 0) {
            remaining = Math.min(remaining, len);
            readBuffer.get(b, off, remaining);

            if (log.isDebugEnabled()) {
                log.debug("Socket: [" + this + "], Read from buffer: [" + remaining + "]");
            }
        }
        return remaining;
    }


    public void close() throws IOException {
        getSocket().close();
    }


    public boolean isClosed() {
        return !getSocket().isOpen();
    }


    private int fillReadBuffer(boolean block) throws IOException {
        socketBufferHandler.configureReadBufferForWrite();
        return fillReadBuffer(block, socketBufferHandler.getReadBuffer());
    }


    private int fillReadBuffer(boolean block, ByteBuffer to) throws IOException {
        int nRead;
        NioChannel channel = getSocket();
        if (block) {
            Selector selector = null;
            try {
                selector = pool.get();
            } catch (IOException x) {
                // Ignore
            }
            try {
               NioSocketWrapper att = (NioSocketWrapper)channel.getAttachment();
                if (att == null) {
                    throw new IOException("Key must be cancelled.");
                }
                nRead = pool.read(to, channel, selector, att.getReadTimeout());
            } finally {
                if (selector != null) {
                    pool.put(selector);
                }
            }
        } else {
            nRead = channel.read(to);
            if (nRead == -1) {
                throw new EOFException();
            }
        }
        return nRead;
    }


    public NioChannel getSocket() {
		return socket;
	}
    
    public final void write(boolean block, byte[] buf, int off, int len) throws IOException {
        if (len == 0 || buf == null) {
            return;
        }

        // While the implementations for blocking and non-blocking writes are
        // very similar they have been split into separate methods to allow
        // sub-classes to override them individually. NIO2, for example,
        // overrides the non-blocking write but not the blocking write.
        if (block) {
            writeBlocking(buf, off, len);
        } else {
            writeNonBlocking(buf, off, len);
        }
    }
    
    /**
     * Transfers the data to the socket write buffer (writing that data to the
     * socket if the buffer fills up using a non-blocking write) until either
     * all the data has been transferred and space remains in the socket write
     * buffer or a non-blocking write leaves data in the socket write buffer.
     *
     * @param buf   The byte array containing the data to be written
     * @param off   The offset within the byte array of the data to be written
     * @param len   The length of the data to be written
     *
     * @throws IOException If an IO error occurs during the write
     */
    protected void writeNonBlocking(byte[] buf, int off, int len) throws IOException {
        if (bufferedWrites.size() == 0 && socketBufferHandler.isWriteBufferWritable()) {
            socketBufferHandler.configureWriteBufferForWrite();
            int thisTime = transfer(buf, off, len, socketBufferHandler.getWriteBuffer());
            len = len - thisTime;
            while (!socketBufferHandler.isWriteBufferWritable()) {
                off = off + thisTime;
                doWrite(false);
                if (len > 0 && socketBufferHandler.isWriteBufferWritable()) {
                    socketBufferHandler.configureWriteBufferForWrite();
                    thisTime = transfer(buf, off, len, socketBufferHandler.getWriteBuffer());
                } else {
                    // Didn't write any data in the last non-blocking write.
                    // Therefore the write buffer will still be full. Nothing
                    // else to do here. Exit the loop.
                    break;
                }
                len = len - thisTime;
            }
        }

        if (len > 0) {
            // Remaining data must be buffered
            addToBuffers(buf, off, len);
        }
    }
    
    protected void addToBuffers(byte[] buf, int offset, int length) {
        ByteBufferHolder holder = getByteBufferHolder(length);
        holder.getBuf().put(buf, offset, length);
    }


    protected void addToBuffers(ByteBuffer from) {
        ByteBufferHolder holder = getByteBufferHolder(from.remaining());
        holder.getBuf().put(from);
    }

    private ByteBufferHolder getByteBufferHolder(int capacity) {
        ByteBufferHolder holder = bufferedWrites.peekLast();
        if (holder == null || holder.isFlipped() || holder.getBuf().remaining() < capacity) {
            ByteBuffer buffer = ByteBuffer.allocate(Math.max(bufferedWriteSize, capacity));
            holder = new ByteBufferHolder(buffer, false);
            bufferedWrites.add(holder);
        }
        return holder;
    }

    /**
     * Writes the data to the socket (writing that data to the socket using a
     * non-blocking write) until either all the data has been transferred and
     * space remains in the socket write buffer or a non-blocking write leaves
     * data in the socket write buffer. If it is possible use the provided
     * buffer directly and do not transfer to the socket write buffer.
     *
     * @param from The ByteBuffer containing the data to be written
     *
     * @throws IOException If an IO error occurs during the write
     */
    protected void writeNonBlocking(ByteBuffer from) throws IOException {
        if (bufferedWrites.size() == 0 && socketBufferHandler.isWriteBufferWritable()) {
            writeNonBlockingInternal(from);
        }

        if (from.remaining() > 0) {
            // Remaining data must be buffered
            addToBuffers(from);
        }
    }


    private boolean writeNonBlockingInternal(ByteBuffer from) throws IOException {
        if (socketBufferHandler.isWriteBufferEmpty()) {
            return writeByteBufferNonBlocking(from);
        } else {
            socketBufferHandler.configureWriteBufferForWrite();
            transfer(from, socketBufferHandler.getWriteBuffer());
            if (!socketBufferHandler.isWriteBufferWritable()) {
                doWrite(false);
                if (socketBufferHandler.isWriteBufferWritable()) {
                    return writeByteBufferNonBlocking(from);
                }
            }
        }

        return !socketBufferHandler.isWriteBufferWritable();
    }


    protected boolean writeByteBufferNonBlocking(ByteBuffer from) throws IOException {
        // The socket write buffer capacity is socket.appWriteBufSize
        int limit = socketBufferHandler.getWriteBuffer().capacity();
        int fromLimit = from.limit();
        while (from.remaining() >= limit) {
            int newLimit = from.position() + limit;
            from.limit(newLimit);
            doWrite(false, from);
            from.limit(fromLimit);
            if (from.position() != newLimit) {
                // Didn't write the whole amount of data in the last
                // non-blocking write.
                // Exit the loop.
                return true;
            }
        }

        if (from.remaining() > 0) {
            socketBufferHandler.configureWriteBufferForWrite();
            transfer(from, socketBufferHandler.getWriteBuffer());
        }

        return false;
    }


    /**
     * Writes as much data as possible from any that remains in the buffers.
     *
     * @param block <code>true</code> if a blocking write should be used,
     *                  otherwise a non-blocking write will be used
     *
     * @return <code>true</code> if data remains to be flushed after this method
     *         completes, otherwise <code>false</code>. In blocking mode
     *         therefore, the return value should always be <code>false</code>
     *
     * @throws IOException If an IO error occurs during the write
     */
    public boolean flush(boolean block) throws IOException {
        boolean result = false;
        if (block) {
            // A blocking flush will always empty the buffer.
            flushBlocking();
        } else {
            result = flushNonBlocking();
        }

        return result;
    }


    protected void flushBlocking() throws IOException {
        doWrite(true);

        if (bufferedWrites.size() > 0) {
            Iterator<ByteBufferHolder> bufIter = bufferedWrites.iterator();
            while (bufIter.hasNext()) {
                ByteBufferHolder buffer = bufIter.next();
                buffer.flip();
                writeBlocking(buffer.getBuf());
                if (buffer.getBuf().remaining() == 0) {
                    bufIter.remove();
                }
            }

            if (!socketBufferHandler.isWriteBufferEmpty()) {
                doWrite(true);
            }
        }

    }
    
    /**
     * Transfers the data to the socket write buffer (writing that data to the
     * socket if the buffer fills up using a blocking write) until all the data
     * has been transferred and space remains in the socket write buffer.
     *
     * @param buf   The byte array containing the data to be written
     * @param off   The offset within the byte array of the data to be written
     * @param len   The length of the data to be written
     *
     * @throws IOException If an IO error occurs during the write
     */
    protected void writeBlocking(byte[] buf, int off, int len) throws IOException {
        // Note: There is an implementation assumption that if the switch from
        //       non-blocking to blocking has been made then any pending
        //       non-blocking writes were flushed at the time the switch
        //       occurred.

        // Keep writing until all the data has been transferred to the socket
        // write buffer and space remains in that buffer
        socketBufferHandler.configureWriteBufferForWrite();
        int thisTime = transfer(buf, off, len, socketBufferHandler.getWriteBuffer());
        while (socketBufferHandler.getWriteBuffer().remaining() == 0) {
            len = len - thisTime;
            off = off + thisTime;
            doWrite(true);
            socketBufferHandler.configureWriteBufferForWrite();
            thisTime = transfer(buf, off, len, socketBufferHandler.getWriteBuffer());
        }
        doWrite(true); 
    }


    /**
     * Write the data to the socket (writing that data to the socket using a
     * blocking write) until all the data has been transferred and space remains
     * in the socket write buffer. If it is possible use the provided buffer
     * directly and do not transfer to the socket write buffer.
     *
     * @param from The ByteBuffer containing the data to be written
     *
     * @throws IOException If an IO error occurs during the write
     */
    protected void writeBlocking(ByteBuffer from) throws IOException {
        // Note: There is an implementation assumption that if the switch from
        // non-blocking to blocking has been made then any pending
        // non-blocking writes were flushed at the time the switch
        // occurred.

        // If it is possible write the data to the socket directly from the
        // provided buffer otherwise transfer it to the socket write buffer
        if (socketBufferHandler.isWriteBufferEmpty()) {
            writeByteBufferBlocking(from);
        } else {
            socketBufferHandler.configureWriteBufferForWrite();
            transfer(from, socketBufferHandler.getWriteBuffer());
            if (!socketBufferHandler.isWriteBufferWritable()) {
                doWrite(true);
                writeByteBufferBlocking(from);
            }
        }
    }
    
    protected void writeByteBufferBlocking(ByteBuffer from) throws IOException {
        // The socket write buffer capacity is socket.appWriteBufSize
        int limit = socketBufferHandler.getWriteBuffer().capacity();
        int fromLimit = from.limit();
        while (from.remaining() >= limit) {
            from.limit(from.position() + limit);
            doWrite(true, from);
            from.limit(fromLimit);
        }

        if (from.remaining() > 0) {
            socketBufferHandler.configureWriteBufferForWrite();
            transfer(from, socketBufferHandler.getWriteBuffer());
        }
    }


    protected boolean flushNonBlocking() throws IOException {
        boolean dataLeft = !socketBufferHandler.isWriteBufferEmpty();

        // Write to the socket, if there is anything to write
        if (dataLeft) {
            doWrite(false);
            dataLeft = !socketBufferHandler.isWriteBufferEmpty();
        }

        if (!dataLeft && bufferedWrites.size() > 0) {
            Iterator<ByteBufferHolder> bufIter = bufferedWrites.iterator();
            while (!dataLeft && bufIter.hasNext()) {
                ByteBufferHolder buffer = bufIter.next();
                buffer.flip();
                dataLeft = writeNonBlockingInternal(buffer.getBuf());
                if (buffer.getBuf().remaining() == 0) {
                    bufIter.remove();
                }
            }

            if (!dataLeft && !socketBufferHandler.isWriteBufferEmpty()) {
                doWrite(false);
                dataLeft = !socketBufferHandler.isWriteBufferEmpty();
            }
        }

        return dataLeft;
    }

    
    private void doWrite(boolean block) throws IOException {
        socketBufferHandler.configureWriteBufferForRead();
        doWrite(block, socketBufferHandler.getWriteBuffer());
    }
    
    
    public void doWrite(boolean block, ByteBuffer from) throws IOException {
        long writeTimeout = getWriteTimeout();
        Selector selector = null;
        try {
            selector = pool.get();
        } catch (IOException x) {
            // Ignore
        }
        try {
            int num = pool.write(from, getSocket(), selector, writeTimeout, block);
            log.debug("write data num " + num);
            if (block) {
                // Make sure we are flushed
                do {
                    if (getSocket().flush(true, selector, writeTimeout)) {
                        break;
                    }
                } while (true);
            }
            updateLastWrite();
        } finally {
            if (selector != null) {
                pool.put(selector);
            }
        }
        // If there is data left in the buffer the socket will be registered for
        // write further up the stack. This is to ensure the socket is only
        // registered for write once as both container and user code can trigger
        // write registration.
    }


    public void registerReadInterest() {
        getPoller().add(getSocket(), SelectionKey.OP_READ);
    }


    public void registerWriteInterest() {
        getPoller().add(getSocket(), SelectionKey.OP_WRITE);
    }



    protected void populateRemoteAddr() {
        InetAddress inetAddr = getSocket().getIOChannel().socket().getInetAddress();
        if (inetAddr != null) {
            remoteAddr = inetAddr.getHostAddress();
        }
    }


    protected void populateRemoteHost() {
        InetAddress inetAddr = getSocket().getIOChannel().socket().getInetAddress();
        if (inetAddr != null) {
            remoteHost = inetAddr.getHostName();
            if (remoteAddr == null) {
                remoteAddr = inetAddr.getHostAddress();
            }
        }
    }


    protected void populateRemotePort() {
        remotePort = getSocket().getIOChannel().socket().getPort();
    }


    protected void populateLocalName() {
        InetAddress inetAddr = getSocket().getIOChannel().socket().getLocalAddress();
        if (inetAddr != null) {
            localName = inetAddr.getHostName();
        }
    }


    protected void populateLocalAddr() {
        InetAddress inetAddr = getSocket().getIOChannel().socket().getLocalAddress();
        if (inetAddr != null) {
            localAddr = inetAddr.getHostAddress();
        }
    }
    
 // --------------------------------------------------------- Utility methods

    private static int transfer(byte[] from, int offset, int length, ByteBuffer to) {
        int max = Math.min(length, to.remaining());
        if (max > 0) {
            to.put(from, offset, max);
        }
        return max;
    }

    private static int transfer(ByteBuffer from, ByteBuffer to) {
        int max = Math.min(from.remaining(), to.remaining());
        if (max > 0) {
            int fromLimit = from.limit();
            from.limit(from.position() + max);
            to.put(from);
            from.limit(fromLimit);
        }
        return max;
    }


    protected void populateLocalPort() {
        localPort = getSocket().getIOChannel().socket().getLocalPort();
    }

	public long getReadTimeout() {
		return readTimeout;
	}

	public void setReadTimeout(long readTimeout) {
		this.readTimeout = readTimeout;
	}

	public boolean isSecure() {
		return secure;
	}

	public void setSecure(boolean secure) {
		this.secure = secure;
	}

	public String getNegotiatedProtocol() {
		return negotiatedProtocol;
	}

	public void setNegotiatedProtocol(String negotiatedProtocol) {
		this.negotiatedProtocol = negotiatedProtocol;
	}

	public String getLocalAddr() {
		return localAddr;
	}

	public void setLocalAddr(String localAddr) {
		this.localAddr = localAddr;
	}

	public String getLocalName() {
		return localName;
	}

	public void setLocalName(String localName) {
		this.localName = localName;
	}

	public int getLocalPort() {
		return localPort;
	}

	public void setLocalPort(int localPort) {
		this.localPort = localPort;
	}

	public String getRemoteAddr() {
		return remoteAddr;
	}

	public void setRemoteAddr(String remoteAddr) {
		this.remoteAddr = remoteAddr;
	}

	public String getRemoteHost() {
		return remoteHost;
	}

	public void setRemoteHost(String remoteHost) {
		this.remoteHost = remoteHost;
	}

	public int getRemotePort() {
		return remotePort;
	}

	public void setRemotePort(int remotePort) {
		this.remotePort = remotePort;
	}

	public int getBufferedWriteSize() {
		return bufferedWriteSize;
	}

	public void setBufferedWriteSize(int bufferedWriteSize) {
		this.bufferedWriteSize = bufferedWriteSize;
	}

	public int getInterestOps() {
		return interestOps;
	}

	public void setInterestOps(int interestOps) {
		this.interestOps = interestOps;
	}

	public void setReadLatch(CountDownLatch readLatch) {
		this.readLatch = readLatch;
	}

	public void setWriteLatch(CountDownLatch writeLatch) {
		this.writeLatch = writeLatch;
	}
	
	

	public void setError(SocketTimeoutException socketTimeoutException) {
//		TODO
	}
    
    

}